package cn.iocoder.yudao.module.forward.mqtt;

//import cn.iocoder.yudao.module.forward.dal.dataobject.datasouce.DataSouceDO;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class MqttManager {
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    public void init(ThreadPoolTaskScheduler threadPoolTaskScheduler) {
        this.threadPoolTaskScheduler = threadPoolTaskScheduler;
    }

    protected Map<String, MqttConnect> clientMaps = new ConcurrentHashMap();
    //为了减少数据库联查增加的map
    protected Map<Integer, String> clientIdName = new ConcurrentHashMap<>();
    protected String tag = "";

    public void createClient(MqttInfo mqttInfo, IMqttCallBack mqttCallback) {
        threadPoolTaskScheduler.submit(new ConnectRunnable(mqttInfo, mqttCallback));
    }

    public MqttConnect getMqttConnect(String name) {
        return clientMaps.get(name);
    }

    public MqttConnect getMqttById(Integer id) {
        try {
            return clientMaps.get(clientIdName.get(id));
        } catch (Exception e) {

        }
        return null;
    }

    public String getMqttNameById(Integer id) {
        return clientIdName.get(id);
    }

    public void doConnect(MqttInfo mqttInfo) {
//        if (mqttInfo == null) {
//            return;
//        }
//        if (clientMaps.containsKey(mqttInfo.getName())) {
//            log.error("内部错误, {}({})已存在", tag, mqttInfo.getName());
//            return;
//        }
//
//        log.info("mqttInfo = {}", mqttInfo);
//        // TODO: 2025/4/22
//        //  回调函数类可单独处理，也可统一处理 单独处理时需要重新实现相关方法
//        MessageHandler handler = null;
//        switch (mqttInfo.getName()) {
//            case "xinjiang":
//                handler = new XinJiangMessageHandler();
//                break;
//            case "sichuan":
//                handler = new SichuanMessageHandler();
//                break;
//            default:
//                handler = new MessageHandlerImpl();
//                break;
//        }
//        if (handler == null) {
//            log.error("{}({})不存在的消息处理器，取消创建", tag, mqttInfo.getName());
//            return;
//        }
//        createClient(mqttInfo, new MqttCallBackImpl(handler, mqttInfo));
    }

    public void closeMqttConnect(String name) {
        try {
            if (clientMaps.containsKey(name)) {
                clientMaps.get(name).close();
                log.info("{}({})关闭", tag, name);
            } else {
                log.info("{}({})不存在", tag, name);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("{}({})关闭失败:{}", tag, name, e.getMessage());
        } finally {
            clientMaps.remove(name);
            for (Iterator<Map.Entry<Integer, String>> it = clientIdName.entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<Integer, String> item = it.next();
                if (!item.getValue().equals(name)) {
                    continue;
                }
                it.remove();
            }
        }
    }

    class ConnectRunnable implements Runnable {
        private MqttInfo mqttInfo;
        private IMqttCallBack mqttCallback;

        public ConnectRunnable(MqttInfo mqttInfo, IMqttCallBack mqttCallback) {
            this.mqttInfo = mqttInfo;
            this.mqttCallback = mqttCallback;
        }

        @Override
        public void run() {
            try {
                if (clientMaps.containsKey(mqttInfo.getName())) {
                    return;
                }
                MqttConnect mqttConnect = new MqttConnect(mqttInfo, mqttCallback);
                mqttConnect.connect();
                clientMaps.put(mqttInfo.getName(), mqttConnect);
                clientIdName.put(mqttInfo.getId(), mqttInfo.getName());
            } catch (Exception e) {
                e.printStackTrace();
                log.error("{}({})创建失败:{}, {}", tag, mqttInfo.getName(), mqttInfo, e.getMessage());
                threadPoolTaskScheduler.schedule(this, new Date(System.currentTimeMillis() + 5 * 1000));
            }
        }
    }

    public String getTag() {
        return tag;
    }

    public MqttInfo subscribe(String name, String topic, Object obj) {
        MqttConnect mqttConnect = clientMaps.get(name);
        if (mqttConnect != null && mqttConnect.isConnected()) {
            mqttConnect.subscribe(topic);
            mqttConnect.getMqttInfo().getTopics().put(topic, obj);
            return mqttConnect.getMqttInfo();
        }
        return null;
    }

    public MqttInfo unSubscribe(String name, String topic, Object obj) {
        MqttConnect mqttConnect = clientMaps.get(name);
        if (mqttConnect != null && mqttConnect.isConnected()) {
            mqttConnect.unsubscribe(topic);
            mqttConnect.getMqttInfo().getTopics().remove(topic);
            return mqttConnect.getMqttInfo();
        }
        return null;
    }
}
